#include "StreamClient.h"
#include "Connector.h"
#include "EventLoop.h"
#include "Manager.h"
#include "PlatLogger.h"
#include "Util.h"
#include "XnccAddress.h"
#include "XnccClient.h"
#include "XnccType.h"
namespace xncc {
namespace foundation {
    StreamClient::StreamClient(EventLoop* loop, std::shared_ptr<XnccAddress> serverAddr)
        : loop_(loop),
          connector_(new Connector(loop, serverAddr)),
          connect_(true),
          nextConnId_(1),
          serverAddrStr_(serverAddr->toStr())
    {
    }

    /// 长期驻留内存的设计
    StreamClient::~StreamClient() = default;

    void StreamClient::init()
    {
        loop_->registerUserFunc(shared::InnerMsgId::ON_PUB_IPC_MSG,
                                [this](const char* msg, uint32_t len, types::thread_id_t tid,
                                       types::connection_id_t connId) { handleIoMsg(msg, len, tid, connId); });
        connector_->setNewConnectionCallback([this](int sockfd) { newConnection(sockfd); });
    }

    void StreamClient::connect()
    {
        // FIXME: check state
        // TRACE_INFO("StreamClient::connect - connecting to %s", connector_->serverAddress().toIpPort().data());
        connect_ = true;
        TRACE_INFO << "engine id:" << loop_->id() << " client start, connect server:"_s << serverAddrStr_;
        connector_->start();
    }

    void StreamClient::disconnect()
    {
        connect_ = false;
        connector_->shutdown(serverChannel_);
    }

    void StreamClient::stop()
    {
        connect_ = false;
        connector_->stop();
    }
    void StreamClient::handleIoMsg(const char* msg, uint32_t len, types::thread_id_t tid, types::connection_id_t connId)
    {
        if (unlikely(serverChannel_ == nullptr)) {
            TRACE_ERROR << "connid disconnected:"_s << connId;
            return;
        }
        if (connIdFd_[connId] != serverChannel_->fd()) {
            TRACE_ERROR << "msg sender set error connid:"_s << connId;
            return;
        }
        serverChannel_->sendMsg(msg, len);
    }

    void StreamClient::newConnection(int sockfd)
    {
        const auto connId = shared::CommonBusi::generateConnId();
        loop_->assertInLoopThread(__FUNCTION__);
        TRACE_INFO << " engine id:"_s << loop_->id() << " connect server success, connid:"_s << connId
                   << " server addr:"_s << serverAddrStr_;
        QUEUE_MGR.pubIPCDataToBusiThread(serverAddrStr_.c_str(), static_cast<uint32_t>(serverAddrStr_.size()),
                                         shared::InnerMsgId::ON_CONNECTED, loop_->id(), connId);

        // 一个连接建立之后，投递连接信息到 关注消息的线程， 然后要把这个fd，放进epool循环中
        /// 这里创建的channel
        /// 是TCP连接的channel，断开连接时需要释放内存的，使用智能指针，但是连接断开的时候，只要erase了这个指针，内存才能被释放
        serverChannel_ = std::make_shared<StreamSocketChannel>(loop_, sockfd, connId);
        serverChannel_->init();
        serverChannel_->setCloseCallback(
            [this, connId](const std::string& reason) { removeConnection(connId, reason); });
        serverChannel_->enableReading();
        connIdFd_[connId] = sockfd;
    }

    void StreamClient::removeConnection(types::connection_id_t connId, const std::string& disconnectedReason)
    {
        if (connIdFd_.find(connId) == connIdFd_.end()) {
            TRACE_ERROR << " engine id:"_s << loop_->id() << " has call disconnected, connid:"_s << connId
                        << " server addr:"_s << serverAddrStr_;
            return;
        }
        TRACE_INFO << " engine id:"_s << loop_->id() << " disconnected, connid:"_s << connId << " server addr:"_s
                   << serverAddrStr_;
        QUEUE_MGR.pubIPCDataToBusiThread(disconnectedReason.c_str(), static_cast<uint32_t>(disconnectedReason.size()),
                                         shared::InnerMsgId::ON_DISCONNECTED, loop_->id(), connId);
        serverChannel_->disableAll();
        serverChannel_->remove();
        serverChannel_.reset();
        connIdFd_.erase(connId);
        // 连接断开了 下一步要进行重连
        if (retry_) {
            connector_->setRetryInterval(retryDelayMs_);
            connector_->restart();
        }
    }
}  // namespace foundation
}  // namespace xncc